package com.codejiwei.flink.source;

import com.alibaba.fastjson2.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.ConnectSchema;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.json.JsonConverter;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.storage.ConverterType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class CustomDebeziumDeserializationSchema implements DebeziumDeserializationSchema<DebeziumRecord> {

    private static final Logger logger = LoggerFactory.getLogger(CustomDebeziumDeserializationSchema.class);
    private static final long serialVersionUID = 1L;
    private transient JsonConverter jsonConverter;
    private final Boolean includeSchema;
    private transient Base64.Decoder decoder;
    private transient ObjectMapper objectMapper;

    public CustomDebeziumDeserializationSchema() {
        this(false);
    }

    public CustomDebeziumDeserializationSchema(Boolean includeSchema) {
        this.includeSchema = includeSchema;
    }

    public void deserialize(SourceRecord record, Collector<DebeziumRecord> out) throws Exception {

        if (this.jsonConverter == null) {
            this.initializeJsonConverter();
        }
        if (this.decoder == null) {
            this.decoder = Base64.getDecoder();
        }
        if (this.objectMapper == null) {
            this.objectMapper = new ObjectMapper();
        }

        String binlog = new String(this.jsonConverter.fromConnectData(record.topic(), record.valueSchema(), record.value()));

        DebeziumRecord binlogJson = objectMapper.readValue(binlog, DebeziumRecord.class);

        try {
            ConnectSchema schema = (ConnectSchema) record.valueSchema();
            for (Field field : schema.fields()) {
                if (field.name().equals("before") || field.name().equals("after")) {
                    JSONObject innerMap = null;
                    if (field.name().equals("before")) {
                        innerMap = binlogJson.getBefore();
                    }
                    if (field.name().equals("after")) {
                        innerMap = binlogJson.getAfter();
                    }
                    if (innerMap == null) {
                        continue;
                    }
                    for (Field valueField : field.schema().fields()) {
                        // Date
                        if (valueField.schema().type() == Schema.Type.INT32 && "io.debezium.time.Date".equals(valueField.schema().name())) {
                            Integer v = jsonObjectGetInteger(innerMap, valueField.name());
                            if (v != null) {
                                Long t = v * 24 * 3600 * 1000L;
                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                                sdf.setTimeZone(TimeZone.getTimeZone("GMT+8:00"));
                                innerMap.put(valueField.name(), sdf.format(t));
                            }
                        }
                        // Time
                        if (valueField.schema().type() == Schema.Type.INT64 && "io.debezium.time.MicroTime".equals(valueField.schema().name())) {
                            Long v = jsonObjectGetLong(innerMap, valueField.name());
                            if (v != null) {
                                long t = v - 8L * 60 * 60 * 1000 * 1000;
                                SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.");
                                sdf.setTimeZone(TimeZone.getTimeZone("GMT+8:00"));
                                innerMap.put(valueField.name(), sdf.format(t/1000) + String.format("%06d", t % 1000000));
                            }
                        }
                        // Year
                        if (valueField.schema().type() == Schema.Type.INT32 && "io.debezium.time.Year".equals(valueField.schema().name())) {
                            Integer v = jsonObjectGetInteger(innerMap, valueField.name());
                            if (v != null) {
                                innerMap.put(valueField.name(), v.toString());
                            }
                        }
                        // DateTime(Timestamp)
                        if (valueField.schema().type() == Schema.Type.INT64 && "io.debezium.time.Timestamp".equals(valueField.schema().name())) {
                            Long v = jsonObjectGetLong(innerMap, valueField.name());
                            if (v != null) {
                                long t = v - 8 * 60 * 60 * 1000;
                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                sdf.setTimeZone(TimeZone.getTimeZone("GMT+8:00"));
                                innerMap.put(valueField.name(), sdf.format(t));
                            }
                        }
                        // DateTime(MicroTimestamp)
                        if (valueField.schema().type() == Schema.Type.INT64 && "io.debezium.time.MicroTimestamp".equals(valueField.schema().name())) {
                            Long v = jsonObjectGetLong(innerMap, valueField.name());
                            if (v != null) {
                                int n = 6;
                                long multiple = (long) Math.pow(10, n);
                                long t = v - 8L * 60 * 60 * multiple;
                                double m = Math.pow(10, n-3);
                                long vt = (long) (t / m);
                                String st = Long.toString(t);
                                int microLength = st.length() - n;
                                String mst = (microLength > 0) ? st.substring(microLength) : "";
                                if ("".equals(mst)) {
                                    mst = "000000";
                                }
                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                sdf.setTimeZone(TimeZone.getTimeZone("GMT+8:00"));
                                String s = sdf.format(vt).concat(".").concat(mst);
                                innerMap.put(valueField.name(), s);
                            }
                        }
                        // TimeStamp
                        if (valueField.schema().type() == Schema.Type.STRING && "io.debezium.time.ZonedTimestamp".equals(valueField.schema().name())) {
                            String v = jsonObjectGetString(innerMap, valueField.name());
                            if (v != null) {
                                v = v.replace("Z", " UTC");
                                SimpleDateFormat sdfUTC = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss Z");
                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                sdf.setTimeZone(TimeZone.getTimeZone("GMT+8:00"));

                                String vConverted;
                                Matcher matcher = Pattern.compile("\\.(.*) UTC$").matcher(v);
                                if (matcher.find()) {
                                    String millions = matcher.group().replace(" UTC", "");
                                    v = v.replace(millions, "");
                                    vConverted = sdf.format(sdfUTC.parse(v).getTime()).concat(millions);
                                } else {
                                    vConverted = sdf.format(sdfUTC.parse(v).getTime());
                                }
                                innerMap.put(valueField.name(), vConverted);
                            }
                        }
                        // Binary
                        if (valueField.schema().type() == Schema.Type.BYTES) {
                            String v = jsonObjectGetString(innerMap, valueField.name());
                            if (v != null) {
                                String vConverted = new String(decoder.decode(v), StandardCharsets.UTF_8);
                                innerMap.put(valueField.name(), vConverted);
                            }
                        }
                        // Bit(1)
                        if (valueField.schema().type() == Schema.Type.BOOLEAN) {
                            Boolean v = jsonObjectGetBoolean(innerMap, valueField.name());
                            if (v != null) {
                                Integer vConverted = v ? 1 : 0;
                                innerMap.put(valueField.name(), vConverted);
                            }
                        }
                        // BigDecimal (如无符号长整形)
                        if (valueField.schema().type() == Schema.Type.BYTES && "org.apache.kafka.connect.data.Decimal".equals(valueField.schema().name())) {
                            Struct valueStruct = (Struct) record.value();
                            Struct opValueStruct = (Struct) valueStruct.get(field.name());
                            BigDecimal v = (BigDecimal) opValueStruct.get(valueField);
                            if (v != null) {
                                String vConverted = v.toString();
                                innerMap.put(valueField.name(), vConverted);
                            }
                        }

                        if (innerMap.get(valueField.name()) == null) {
                            // Remove Null value field
                            innerMap.remove(valueField.name());
                        } else {
                            // Convert to String
                            if (valueField.schema().type() != Schema.Type.STRING) {
                                innerMap.put(valueField.name(), innerMap.getString(valueField.name()));
                            }
                        }
                    }
                }
            }

            // 获取pk
            binlogJson.setPkNames(new ArrayList<>());
            binlogJson.setPkName("");
            binlogJson.setPkValue("");
            Struct keyStruct = (Struct) record.key();
            if (keyStruct != null) {
                Schema keySchema = keyStruct.schema();
                if (keySchema != null) {
                    List<Field> fieldList = keySchema.fields();
                    if (fieldList != null && fieldList.size() > 0) {
                        String pkField = fieldList.get(0).name();
                        binlogJson.setPkName(pkField);
                        String type = binlogJson.getOp();
                        JSONObject fieldsMap = ("d".equals(type)) ? binlogJson.getBefore() : binlogJson.getAfter();
                        if (fieldsMap != null) {
                            binlogJson.setPkValue(fieldsMap.getString(pkField));
                        }
                        ArrayList<String> pkNamesList = new ArrayList<>();
                        for (Field field : fieldList) {
                            pkNamesList.add(field.name());
                        }
                        binlogJson.setPkNames(pkNamesList);
                    }
                }
            }

        } catch (Exception e) {
            logger.error("schema convert failed, binlog={}", binlog);
            throw e;
        }
        out.collect(binlogJson);
    }

    private Integer jsonObjectGetInteger(JSONObject innerMap, String key) {
        return innerMap.getInteger(key);
    }

    private Long jsonObjectGetLong(JSONObject innerMap, String key) {
        return innerMap.getLong(key);
    }

    private String jsonObjectGetString(JSONObject innerMap, String key) {
        return innerMap.getString(key);
    }

    private Boolean jsonObjectGetBoolean(JSONObject innerMap, String key) {
        return innerMap.getBoolean(key);
    }

    private void initializeJsonConverter() {
        this.jsonConverter = new JsonConverter();
        HashMap<String, Object> configs = new HashMap<>(2);
        configs.put("converter.type", ConverterType.VALUE.getName());
        configs.put("schemas.enable", this.includeSchema);
        this.jsonConverter.configure(configs);
    }

    public TypeInformation<DebeziumRecord> getProducedType() {
        return BasicTypeInfo.of(DebeziumRecord.class);
    }
}